package me.zhengjie.gas.mqtt;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.util.StdDateFormat;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttClient;
import lombok.extern.slf4j.Slf4j;
import me.zhengjie.gas.dto.TransactionDataDto;
import me.zhengjie.gas.mqtt.config.MqttClientProperties;
import me.zhengjie.gas.service.DtuDeviceService;
import me.zhengjie.gas.service.DtuGpsLogService;
import me.zhengjie.gas.service.TransactionService;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author tsg
 * @version 1.0
 * @description: MqttTemplate
 * @date 2022/4/28 11:07
 */
@Slf4j
public class MqttTemplate {

	private GenericObjectPool<MqttClient> clientPool;

    private ObjectMapper objectMapper = new ObjectMapper();

	@Autowired
	private TransactionService transactionService;

	@Autowired
	private DtuGpsLogService dtuGpsLogService;

	@Autowired
	private DtuDeviceService dtuService;

	private static MqttTemplate instance;

	public static MqttTemplate getInstance() {
		return instance;
	}

	@PostConstruct
	public void init() {
		objectMapper.registerModule(new JavaTimeModule());
		objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
		objectMapper.setDateFormat(new StdDateFormat().withColonInTimeZone(false));
		objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);

		// 自定义 LocalDateTime 反序列化格式
		objectMapper.configOverride(LocalDateTime.class)
				.setFormat(JsonFormat.Value.forPattern("yyyy-MM-dd HH:mm:ss"));
	}

	public static boolean publish(String topic, String content) {
		if (instance == null)
			return false;

		return instance.publish(topic, Buffer.buffer(content), MqttQoS.AT_MOST_ONCE, true, false);
	}

	public static boolean publish(String topic, byte[] bytes) {
		if (instance == null)
			return false;


		return instance.publish(topic, Buffer.buffer(bytes), MqttQoS.AT_MOST_ONCE, true, false);
	}


	public MqttTemplate(MqttClientFactory factory, MqttClientProperties properties) {
		GenericObjectPoolConfig<MqttClient> config = new GenericObjectPoolConfig<>();
		config.setMinIdle(properties.getMinIdle());
		config.setBlockWhenExhausted(properties.isBlockWhenExhausted());
		// 一定要打开,因为创建连接客户端是异步的,需要在获取的使用对客户端进行判断
		config.setTestOnBorrow(properties.isTestOnBorrow());
		config.setTestOnReturn(properties.isTestOnReturn());
		config.setTestWhileIdle(properties.isTestWhileIdle());
		config.setMaxIdle(properties.getMaxIdle());
		config.setMaxTotal(properties.getMaxTotal());
		//一定要关闭jmx，不然springboot启动会报已经注册了某个jmx的错误
		config.setJmxEnabled(properties.isJmxEnabled());
		this.clientPool = new GenericObjectPool<>(factory, config);
		//这里可以做一些初始化连接
		subscribe(properties.getSubscribeTopic(), properties.getQosLevel());
		instance = this;
	}

	public boolean publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain) {
		try {
			MqttClient client = clientPool.borrowObject();
			if (client.isConnected()) {
				log.info("{}获取连接成功", log.getName());
				client.publish(topic, payload, qosLevel, isDup, isRetain, r -> {
					log.info("{}消息推送成功", log.getName());
					// 归还客户端
					clientPool.returnObject(client);
				});
				return true;
			} else {
				log.error("{}获取的客户端是断开的！", log.getName());
			}
		} catch (Exception e) {
			log.error("{}获取连接失败:{}", log.getName(), e.getMessage());
		}
		return false;
	}

	public boolean subscribe(String topic, int qosLevel) {
		try {
			MqttClient client = clientPool.borrowObject();

			if (client.isConnected()) {
				log.info("{}获取连接成功", log.getName());
				client.publishHandler(s -> {
					received(s.topicName(), s.payload(), s.qosLevel(), s.isDup(), s.isRetain());
				});
				client.subscribe(topic, qosLevel, r -> {
					log.info("{}消息订阅成功", topic);
					// 归还客户端
					clientPool.returnObject(client);
				});

				return true;
			}
		} catch (Exception e) {
			log.error("{}获取连接失败:{}", log.getName(), e.getMessage());
		}
		return false;
	}

	/**
	 * 接收到MQTT信息
	 *
	 * @param topic
	 * @param payload
	 * @param qosLevel
	 * @param isDup
	 * @param isRetain
	 */
	public void received(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain) {
		Pattern transaction = Pattern.compile("^\\/GAS\\/.*\\/transaction$");
		Pattern gps = Pattern.compile("^\\/GAS\\/DTU\\/([0-9a-fA-F]{12})\\/SEND$");
		Pattern oilSupply = Pattern.compile("^\\/GAS\\/SUPPLY\\/([0-9a-fA-F]{12})\\/NEW$");

		String content = payload.toString();
		if (transaction.matcher(topic).find()) {
			onTransaction(content);
		} else if (gps.matcher(topic).find()) {
			Matcher matcher = gps.matcher(topic);
			if (matcher.find()) {
				String deviceId = matcher.group(1);
				byte[] bytes = payload.getBytes();
				if (bytes[0] == (byte)0xFA) {
					byte[] reply = dtuService.handleBytes(deviceId, bytes);
					if (reply != null) {
						// 回复信息
						publish("/GAS/DTU/" + deviceId + "/RECV", reply);
					}
				} else {
					onDtuGpsMessage(deviceId, content);
				}
			}
		} else if (oilSupply.matcher(topic).find()) {
			Matcher matcher = oilSupply.matcher(topic);
			if (matcher.find()) {
				String deviceId = matcher.group(1);
				byte[] bytes = payload.getBytes();
			}
		}

		log.info("{}收到消息:{}", topic, content);
	}

	private void onTransaction(String content) {
		if (!content.startsWith("FA")) {
			try {
				TransactionDataDto dto = objectMapper.readValue(content, TransactionDataDto.class);
				transactionService.newTransaction(dto);
			} catch (JsonProcessingException e) {
				e.printStackTrace();
			}
		}
	}

	private void onDtuGpsMessage(String deviceId, String content) {
		try {
			Map map = objectMapper.readValue(content, Map.class);
			dtuGpsLogService.onRecveive(deviceId, map);
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
	}

}
